package ranks.tasks.calc;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Iterator;

import mapred.example.base.TextReducer;
import mapred.util.FeatureMap;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import ranks.Customize;
import ranks.Customize.DataType;
import ranks.base.Const;
import ranks.base.Data;

public class Calc_Reducer extends TextReducer {

	@Override
	public void reduce(Text key, Iterator<Text> values,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		Data rank = new Data();
		Data old = new Data();
		FeatureMap map = FeatureMap.createNewFeatureMap(Const.DELIM);
		while (values.hasNext()) {
			Text value = values.next();
			String text = value.toString().trim();
			if (text.startsWith(Const.RANK)) {
				rank.add(new Data(text.substring(Const.RANK.length())));
			} else if (text.startsWith(Const.OLD)) {
				old = new Data(text.substring(Const.OLD.length()));
			} else {
				map = FeatureMap.parse(text, Const.DELIM);
			}
		}
		
		rank.set(DataType.BIAS, old.get(DataType.BIAS));
		rank.set(DataType.OUT_DEGREE, old.get(DataType.OUT_DEGREE));

		if ("Z".equals(key.toString())) {
			JobConf conf = this.getJobConf();
			BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(
					FileSystem.get(conf).create(
							new Path(this.getJobConfConfig("Output") + "/Z_"
									+ this.getJobConfConfig("Round")), true)));
	        long websize = Long.parseLong(this.getJobConfConfig("WebSize"));
			bw.write(Customize.fomularZ(rank, websize).toString());
			bw.flush();
			bw.close();
		} else {
			rank = Customize.fomular(rank);
			output.collect(key, new Text(rank + "\t" + map.toString()));
			reporter.incrCounter("WebSize", "WebSize", 1);
			Customize.delta(rank, old, reporter); 
		}
	}

	@Override
	public void setUp() {
	}

}
